package fnsync
import java.lang
import java.util.concurrent.ConcurrentHashMap

import com.googlecode.scalascriptengine.EvalCode
import javax.cache.event.{CacheEntryEvent, CacheEntryEventFilter, CacheEntryUpdatedListener, EventType}
import org.apache.ignite.cache.CacheMode
import org.apache.ignite.{IgniteCache, Ignition}
import org.apache.ignite.cache.query.ContinuousQuery
import org.apache.ignite.configuration.{CacheConfiguration, IgniteConfiguration}
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi
import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder
import ws.very.util.lang.SingleObj

import scala.collection.convert.WrapAsScala._
import scala.reflect.ClassTag
import scala.reflect.runtime.universe
import scala.util.Try
import scala.reflect.runtime.universe._
import scala.compat.java8.FunctionConverters._
import ignite.lang.FnConverter._
import javax.cache.configuration.Factory
object FilterFac extends Factory[CacheEntryEventFilter[String,String]] {
  override def create(): CacheEntryEventFilter[String, String] = toCacheEntryEventFilter(_=>true)
}

trait FnSyncer[R] {
  implicit def rType: TypeTag[R]


  def codeCache: IgniteCache[String, String]




  private val evalCodeCache = new ConcurrentHashMap[String, R]

  {
    val qry = new ContinuousQuery[String, String]
    qry.setInitialQuery(Query.Scan.all[String,String])
    qry.setRemoteFilterFactory(FilterFac)
    qry.setLocalListener(toCacheEntryUpdatedListener[String, String] { itb =>
       itb.foreach({ item =>

        item.getEventType match {
          case EventType.EXPIRED | EventType.REMOVED => evalCodeCache.remove(item.getKey)
          case EventType.CREATED | EventType.UPDATED if item.getValue != item.getOldValue =>
            evalCodeCache.put(item.getKey, EvalCode.newObj[R](item.getValue))

        }
        //println(evalCodeCache)
      })
    })

    codeCache.query(qry).getAll.foreach{item=>
      evalCodeCache.put(item.getKey, EvalCode.newObj[R](item.getValue))
    }
  }


  def regScalaFnByName(name: String, code: String) = {
    val nName = name.trim
    val nCode = code.trim
    assert(nName.nonEmpty && nCode.nonEmpty)
    Try {
      EvalCode.newObj[R](nCode)
    }.map { r =>
      codeCache.put(nName, nCode)

    }

  }

  def getScalaFnByName(name: String) = evalCodeCache.get(name.trim)



  def remove(name: String) = codeCache.remove(name.trim)

}

object FnSyncer {
   val ignite = Ignition.getOrStart({
    val conf = new IgniteConfiguration
    conf.setDiscoverySpi((new TcpDiscoverySpi).setIpFinder((new TcpDiscoveryMulticastIpFinder).setMulticastGroup("238.238.238.238")))

    conf
  })

  protected class FnSyncerImpl[R: TypeTag](cacheName: String) extends FnSyncer[R] {

    override lazy val codeCache: IgniteCache[String, String] = {
      ignite.getOrCreateCache((new CacheConfiguration[String, String]).
        setCacheMode(CacheMode.REPLICATED).setName(cacheName))
    }

    override  def rType: universe.TypeTag[R] = implicitly
  }

  private val nameSyncMapping = new ConcurrentHashMap[String, FnSyncerImpl[_]]

  def getOrCreate[R: TypeTag](cacheName: String) = nameSyncMapping.computeIfAbsent(cacheName, { key: String =>
    new FnSyncerImpl[R](cacheName)

  }.asJava)

}


object TestWork extends App {
  FnSyncer.getOrCreate[() => Int]("()=>Int").regScalaFnByName("2", "()=>2")
  //println(FnSyncer.getOrCreate[() => Int]("()=>Int").getScalaFnByName("3").asInstanceOf[()=>Int]())
  //FnSyncer.getOrCreate[() => Int]("()=>Int").remove("8")
  println(FnSyncer.ignite.getOrCreateCache("()=>Int").foreach(println))
}
